-module(rtps_reader_locator).

%%% @doc With every stateless writer, a reader locator is associated
%%% which represents the channel used by the writer for sending
%%% changes to remote readers. Because a writer can use one or more
%%% channels in parallel, the reader locators are implemented as
%%% separate processes. The type of locator used determines many of
%%% the characteristics of the reader locator, such as time-outs and
%%% periodic events.
%%%
%%% The reader locator is implemented as a state machine with three
%%% parallel states.
%%%
%%% TODO: Review https://issues.omg.org/issues/DDSIRTP23-100

%%% TODO: do something with nackSuppressionDuration

%%% TODO: Sending a HeartBeat message when there is no data is
%%% unspecified see https://issues.omg.org/issues/DDSIRTP23-66

%%% TODO: When does Best-Effort Stateless Writer send a GAP: see
%%% https://issues.omg.org/issues/DDSIRTP23-21. GAP messages are used
%%% for "irrelevant' data samples, but a reader locator doesn't filter
%%% on behalf of a reader because it is unaware which readers are
%%% tuning in, nor can it distinguish between them. However, there is a
%%% TIME_BASED_FILTER QoS thingy which may cause GAPs?

-behaviour(gen_statem).

%% -include_lib("kernel/include/logger.hrl").
-include("rtps.hrl").

%% API
-export([start_link/5, add_change/3]). 

%% gen_fsm callbacks
-export([init/1, callback_mode/0, handle_event/4, terminate/3, code_change/4]).

%% Since the reader locator has no knowledge of which readers are
%% listening, the reader locator uses the UNKNOWN Guid for addressing
%% the reader.

%% TODO: A reader locator could also be configured to 'broadcast' to
%% specific readers by not using the GUID `unknown', but set the GUID
%% prefix to `unknown' and the entity id to the id of the readers that
%% are supposed to receive the data. Sending data to a specific
%% reader, with a fully qualified GUID, is done using a reader proxy
%% and not a reader locator.
-define(READER_GUID_UNKNOWN, #guid{guid_prefix = unknown, entity_id = unknown}).

%% NB: there is actually little need for keeping state1. With
%% best_effort state1 becomes idle and with reliable, it is always
%% announcing.
-record(state, {reliability_level :: best_effort | reliable,
		% state1 :: idle | announcing, % | pushing,
		state2 :: undefined | waiting | must_repair}). % | repairing
%% See table 8.51
-record(data, {locator :: rtps:locator(), writer_guid :: rtps:guid(), writer_cache :: pid(),
	       push_mode :: boolean() | undefined,
	       requested_changes :: [rtps:sequence_number()] | undefined,
	       expects_inline_qos :: boolean(), inline_qos :: undefined | [rtps_qos:qos()],
	       heartbeat_period :: rtps:duration() | undefined, 
	       heartbeat_count :: pos_integer() | undefined,
	       nack_response_delay :: rtps:duration() | undefined,
	       transport_pid :: pid()}).

%%%===================================================================
%%% API
%%%===================================================================

start_link(Locator, Reg, Writer_guid, Writer_cache, Opts) ->
    gen_statem:start_link({via, rtps_reg, {Reg, Locator}}, ?MODULE, [Locator, Writer_guid, Writer_cache, Opts], []).

-spec add_change(Pid, [Seq_num], Qos) -> ok when
      Pid :: pid(),
      Seq_num :: rtps:sequence_number(),
      Qos :: undefined | [rtps_qos:qos()].
%% @doc This operation adds the CacheChange(s) to the
%% unsent_changes. The CacheChange(s) are referred to by their
%% sequence number and fetched from the HistoryCache as needed.
add_change(Pid, Seq_nums, Qos) ->
    gen_statem:call(Pid, {add_change, Seq_nums, Qos}).

%%%===================================================================
%%% gen_statem callbacks
%%%===================================================================

%% @private
init([Locator, Writer_guid, Writer_cache, Opts]) ->
    Expects_inline_qos = proplists:get_value(expects_inline_qos, Opts, true),
    Reliability_level = proplists:get_value(reliability_level, Opts),
    %% Connect to the transport. This will return the pid of the
    %% sender process associated with this transport.
    Transports_sup = proplists:get_value(transports_sup, Opts),
    Data = #data{locator = Locator, writer_guid = Writer_guid, writer_cache = Writer_cache,
    		 expects_inline_qos = Expects_inline_qos},
    case Reliability_level of
    	best_effort ->
	    %% A best effort reader locator (as part of a stateless,
	    %% best-effort writer) has no defined remote guids, not
	    %% even the guid 'unknown'. If 'unknown' was used instead
	    %% of 'undefined' in the next call as the remote guid, the
	    %% RTPS receiver will consider this endpoint as a
	    %% candidate for incoming messages, which is not correct
	    %% since the best effort reader locator never ever will
	    %% process any incoming messages.
	    {ok, Pid} = rtps_transport_sup:add(Transports_sup, self(), Locator, Writer_guid, undefined, Opts),
    	    {ok, #state{reliability_level = best_effort},
	     Data#data{transport_pid = Pid}};
    	reliable ->
	    {ok, Pid} = rtps_transport_sup:add(Transports_sup, self(), Locator, Writer_guid, unknown, Opts),
	    Push_mode = proplists:get_value(push_mode, Opts, true),
	    Heartbeat_period = proplists:get_value(heartbeat_period, Opts, ?HEARTBEAT_PERIOD),
	    Nack_response_delay = proplists:get_value(nack_response_delay, Opts, ?NACK_RESPONSE_DELAY),
    	    {ok, #state{reliability_level = reliable, state2 = waiting},
    	     Data#data{transport_pid = Pid, requested_changes = [], push_mode = Push_mode,
    		       heartbeat_period = Heartbeat_period, heartbeat_count = 1,
    		       nack_response_delay = Nack_response_delay},
    	     {{timeout, heartbeat}, 0, undefined}}
    end.

%% @private
callback_mode() ->
    handle_event_function.

%% @private

%% ----- Events from the API 

%% For a reliable reader locator, sending DATA samples should be
%% concluded by sending a HEARTBEAT. In some implementations, the
%% exact behavior of the HEARTBEATs is configurable. Sending a
%% HEARTBEAT will also every time reset the heartbeat timer.

%% T2 unsent_changes() != <empty>
handle_event({call, From}, {add_change, Seq_nums, Qos},
	     #state{reliability_level = Reliability_level},
	     #data{writer_guid = Writer_guid, writer_cache = Writer_cache,
		   push_mode = Push_mode, expects_inline_qos = Expects_inline_qos,
		   heartbeat_period = Heartbeat_period, heartbeat_count = Count,
		   transport_pid = Transport_pid} = Data) ->
    Inline_qos = case Expects_inline_qos of
		     true ->
			 Qos;
		     false ->
			 undefined
		 end,
    case Reliability_level of
	best_effort ->
	    Cc_msgs = cc_msgs(Writer_cache, Writer_guid, Seq_nums, Inline_qos),
	    rtps_sender:send(Transport_pid, Cc_msgs),
	    {keep_state, Data#data{inline_qos = Inline_qos}, [{reply, From, ok}]};
	reliable ->
	    Cc_msgs = case Push_mode of
			  true ->
			      cc_msgs(Writer_cache, Writer_guid, Seq_nums, Inline_qos);
			  false ->
			      []
		      end,
	    {First_sn, Last_sn} = rtps_history_cache:get_seq_num_min_max(Writer_cache, Writer_guid),
	    Heartbeat = heartbeat(Writer_guid, First_sn, Last_sn, Count),
	    rtps_sender:send(Transport_pid, Cc_msgs ++ [Heartbeat]),
	    {keep_state, Data#data{inline_qos = Inline_qos, heartbeat_count = Count + 1},
	     [{reply, From, ok}, {{timeout, heartbeat}, Heartbeat_period, undefined}]}
    end;

%% ----- Events from the receiver

%% We can receive a bunch of submessages from the receiver but want to
%% process them one by one, because this is a state machine and each
%% submessage determines state. NB: a best effort reader locator will
%% never process any incoming messages.
handle_event(info, {rcv, Rcvs}, #state{reliability_level = reliable}, _Data) ->
    {keep_state_and_data, [{next_event, internal, Rcvs}]};
handle_event(internal, [], _State, _Data) ->
    keep_state_and_data;

%% T6 / T8 ACKNACK message is received. Note that the processing of
%% this message uses the reply locators in the RTPS Receiver as
%% stipulated out in the specs!
handle_event(internal, 
	     [#msg{submsg = #acknack{final_flag = _Final_flag, 
				     reader_sn_state = #sequence_number_set{set = Seq_nums}},
		   reply_to = Reply_to} | Rcvs],
	     #state{reliability_level = reliable, state2 = State2} = State,
	     #data{locator = Locator, requested_changes = Requested_changes,
		   nack_response_delay = Nack_response_delay} = Data) ->
    case lists:member(Locator, Reply_to) of
	true ->
	    case State2 of
		must_repair ->
		    Requested_changes1 = lists:umerge(Requested_changes, Seq_nums),
		    {keep_state, Data#data{requested_changes = Requested_changes1},
		     [{next_event, internal, Rcvs}]};
		waiting when Seq_nums =/= [] ->
		    {next_state, State#state{state2 = must_repair},
		     Data#data{requested_changes = Seq_nums},
		     [{{timeout, nack_response_delay}, Nack_response_delay, undefined},
		      {next_event, internal, Rcvs}]};
		waiting ->
		    {keep_state_and_data, [{next_event, internal, Rcvs}]}
	    end;
	false ->
	    {keep_state_and_data, [{next_event, internal, Rcvs}]}
    end;

%% ----- Time-out Events

%% T5 Heartbeat timer expires.
handle_event({timeout, heartbeat}, _Content, _State,
	     %% #state{reliability_level = reliable} = _State, % Not needed
	     #data{writer_cache = Writer_cache, writer_guid = Writer_guid,
		   heartbeat_period = Heartbeat_period, heartbeat_count = Count,
		   transport_pid = Transport_pid} = Data) ->
    case rtps_history_cache:get_seq_num_min_max(Writer_cache, Writer_guid) of
	{First_sn, Last_sn} ->
	    Heartbeat = heartbeat(Writer_guid, First_sn, Last_sn, Count),
	    rtps_sender:send(Transport_pid, Heartbeat),
	    {keep_state, Data#data{heartbeat_count = Count + 1},
	     [{{timeout, heartbeat}, Heartbeat_period, heartbeat}]};
	_ ->
	    {keep_state_and_data, [{{timeout, heartbeat}, Heartbeat_period, heartbeat}]}
    end;
%% T9 Nack response delay expires, start reparing
handle_event({timeout, nack_response_delay}, _Content,
	     #state{state2 = must_repair} = State,
	     #data{writer_cache = Writer_cache, writer_guid = Writer_guid,
		   requested_changes = Requested_changes, inline_qos = Inline_qos,
		   heartbeat_period = Heartbeat_period, heartbeat_count = Count,
		   transport_pid = Transport_pid} = Data) ->
    Cc_msgs = cc_msgs(Writer_cache, Writer_guid, Requested_changes, Inline_qos),
    {First_sn, Last_sn} = rtps_history_cache:get_seq_num_min_max(Writer_cache, Writer_guid),
    Heartbeat = heartbeat(Writer_guid, First_sn, Last_sn, Count),
    rtps_sender:send(Transport_pid, Cc_msgs ++ [Heartbeat]),
    {next_state, State#state{state2 = waiting},
     Data#data{requested_changes = [], heartbeat_count = Count + 1},
     [{{timeout, heartbeat}, Heartbeat_period, undefined}]};

%% TODO: Sink unhadled events. Remove these clauses.
handle_event({call, From} = _Event_type, _Event_content, _State, _Data) ->
    %% ?LOG_DEBUG("Unhandled event: Type ~p, Content ~p", [Event_type,  Event_content]),
    Reply = ok,
    {keep_state_and_data, [{reply, From, Reply}]};
handle_event(_Event_type,  _Event_content, _State, _Data) ->
    %% ?LOG_DEBUG("Unhandled event: Type ~p, Content ~p", [Event_type,  Event_content]),
    keep_state_and_data.

%% @private
terminate(_Reason, _state, _Data) ->
    ok.

%% @private
code_change(_Old_vsn, state, Data, _Extra) ->
    {ok, state, Data}.

%%%===================================================================
%%% Internal functions
%%%===================================================================

cc_msgs(Writer_cache, Writer_guid, Seq_nums, Inline_qos) ->
    %% Because a reader locator does not filter on behalf of remote
    %% readers, there are no irrelevant cache changes and therefor
    %% there are no GAP messages. This in contrast to the reader
    %% proxy.
    rtps_submsg:data(Writer_guid, ?READER_GUID_UNKNOWN, Writer_cache, Seq_nums, Inline_qos).

heartbeat(Writer_guid, First_sn, Last_sn, Count) ->
    Final_flag = true, % true, in contrast with reader proxy
    Liveliness_flag = false,
    rtps_submsg:heartbeat(Writer_guid, ?READER_GUID_UNKNOWN, First_sn, Last_sn,
			  Count, Final_flag, Liveliness_flag).
	    
    
